package org.kth.dks.dks_comm;

import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.log4j.Logger;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_marshal.BootstrapMsg;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.MsgSrcDestWrapper;

class ConnHandlerOut extends Thread {
	private Logger log = Logger.getLogger(ConnHandlerOut.class);
	private final int MAXSEND	  		  = 25;
	private DKSNetAddress destNode        = null;
	private List msgQueue                 = Collections.synchronizedList(new LinkedList());
	private Map unackedMsgs               = Collections.synchronizedMap(new HashMap());
	private final int SOCKETTIMEOUT       = 0;     //No timeout!
	private boolean finish                = false;
	private ConnectionHandler connHandler = null;
	private DKSMarshal marshal;
	
	private Socket socket                   = null;
	DataOutputStream outStream              = null;
	private double rtt                      = DEFAULTRTT;
	private final static double DEFAULTRTT  = 1000.0d;
	private final static double RTOFACTOR   = 7.0;
	private final static double MAXRTO      = 20000.0;
	private final static double MINRTO      = 1000.0;
	
	private int msgNo                     = 0;
	
	private final Timer rttTimer;
	
	private final class TimeoutDetector extends TimerTask {
		
		public void run() {
			if (finish && unackedMsgs.isEmpty())
				rttTimer.cancel();
			
			synchronized (unackedMsgs) {
				for (Iterator it = unackedMsgs.values().iterator(); it.hasNext(); ) {
					UnackedMsg msg = (UnackedMsg) it.next();
					double elapsed = (new Date()).getTime() - msg.rttStart;
					if (elapsed > MINRTO && 
							(elapsed > ((double) RTOFACTOR)*rtt || elapsed>MAXRTO)
					) {
						it.remove();
						log.warn(
								"Timeout after " + elapsed + "msec with avg rtt=" + rtt +
								" on socket " + destNode);
						marshal.failureHandler(msg.msg);
					}
				}
			}
		}
	}
	
	private final TimeoutDetector timeDetect = new TimeoutDetector();
	
	// I've choosen a simple construction for
	// storing messages because messages are only
	// stored locally.
	class OutMsgElement{
		public ConnMessageTypes type;
		public Object load;
		public DKSNetAddress src;
		public OutMsgElement(DKSNetAddress _src, ConnMessageTypes type, Object load){
			src = _src;
			this.type = type;
			this.load = load;
		}
	}
	
	class UnackedMsg{
		private MsgSrcDestWrapper msg;
		private long rttStart = 0;
		
		UnackedMsg(MsgSrcDestWrapper msg){
			this.msg = msg;
			rttStart = (new Date()).getTime();
		}
		
		long calculateRTT() {
			long rttEnd = (new Date()).getTime();
			long newRTT = rttEnd - rttStart;
			rttStart = 0;
			return newRTT;
		}
	}
	
	
	public ConnHandlerOut(ConnectionHandler ch, DKSNetAddress src, DKSNetAddress dest) throws DKSRefNoResponse
	{
		log.info("Creating new sender -- connecting "+dest);
		this.connHandler = ch;
		this.marshal     = connHandler.getDKSMarshal();
		this.destNode = dest;
		
		try {
			establishConnection();
		} catch (IOException ex) {
			throw new DKSRefNoResponse(dest);
		}
		
		addElement(new OutMsgElement(src, ConnMessageTypes.PRESENT_MSG ,null));
		this.setName(ConnHandlerOut.class.getName() + "(" + dest + ")");
		this.start();
		rttTimer = new Timer(ConnHandlerOut.class.getName() + ".RTTTimer");
		rttTimer.schedule(timeDetect, 0, 4000);
	}
	
	public ConnHandlerOut(ConnectionHandler ch, DKSNetAddress dest, Socket soc){
		log.info("Creating new sender -- accepting "+dest);
		this.connHandler=ch;
		this.destNode = dest;
		this.socket = soc;
		try {
			acceptConnection();
		} catch (IOException ex) { log.error( "Couldn't create connection to "+dest+"\n"+ex); }
		this.setName(ConnHandlerOut.class.getName() + "(" + dest + ")");
		this.start();
		rttTimer = new Timer(ConnHandlerOut.class.getName() + ".RTTTimer");
		rttTimer.schedule(timeDetect, 0, 4000);
	}
	
	public Socket getSocket() { return socket; }
	
	public void sendMessage(DKSNetAddress src, Object msg ){
		log.info("Sending msg  to "  + destNode);
		addElement(new OutMsgElement(src, ConnMessageTypes.CONTENTS_MSG, msg));
		
	}
	
	public void sendAck(int msgId){
		log.info("Sending ack " + msgId + " to "  + destNode);
		addElement(new OutMsgElement(null, ConnMessageTypes.ACK_MSG, new Integer(msgId)));
	}
	
	
	public void ackReceived(int msgId){
		// This method removes a message in the out queue
		// beacuse the message has been properly delivered.
		// Moreover, the rtt is calsualted.
		//
		// Note that the rtt is not used yet!
		log.info("Received ack " + msgId + " from "  + destNode);
		
		UnackedMsg msg = (UnackedMsg) unackedMsgs.remove(new Integer(msgId));
		if (msg != null) {
			
			long newRTT = msg.calculateRTT();
			if( newRTT > 0 ) {
				double alpha = 0.8;
				rtt = rtt * alpha + ((double)newRTT) * (1.0 - alpha);
				log.debug("AVG=" + rtt + " NEW=" + (double)newRTT);
			}
		}
		else{
			log.warn( "Received an ack for a message that does not exists :"+msgId);
			
		}
	}
	
	public int getUnackedMsgs(){
		return unackedMsgs.size();
	}
	
	synchronized void end() {
		finish = true;
		notifyAll();
	}
	
	public void run() {
		// potentially we could make connection establishment
		// asynchrounous as well. Thus, there is one thread
		// that handles out-bound trafic. This thread starts
		// a dedicated thread that estabshes a connection.
		
		// That schema would (potentially) be able to handle
		// simultanous connection atempts in a nice way.
		log.info("Creating sender " + destNode);
		
		while (!finish ){
			deliverMessage();
			Thread.yield();
		}
		log.info("Closing ConnHandlerOut to node "+destNode);
	}
	
	/* This method sends the message in byte format.
	 */
	private synchronized void deliverMessage() {
		
		OutMsgElement msgEle = getNextElement();
		if (msgEle==null) return;
		
		log.info("Delivering message of type " +  msgEle.type + " to "  + destNode);
		
		try {
			// msgformat <TYPE><LENGTH><LOAD>
			if (msgEle.type==ConnMessageTypes.CONTENTS_MSG) {
				msgNo++;
				MsgSrcDestWrapper mesg = (MsgSrcDestWrapper) msgEle.load;
				unackedMsgs.put(new Integer(msgNo), new UnackedMsg(mesg)); // clock is read here
			}
			
			outStream.writeByte(msgEle.type.toByte());
			
			if (msgEle.type==ConnMessageTypes.CONTENTS_MSG) {
				MsgSrcDestWrapper mesg = (MsgSrcDestWrapper) msgEle.load;
				outStream.writeByte(DKSMarshal.TRANSDEFAULT);
				outStream.writeInt(msgNo);
				byte []out_data = marshal.marshalMsgSrcDestWrapper(mesg);
				ConnMessageMiscs.m_writeDataBlock(out_data, outStream);
				
				connHandler.statAddBytesSent(out_data.length);
				log.info("msg " + msgNo  + " is sent" );
			} else if (msgEle.type==ConnMessageTypes.PRESENT_MSG) {
				BootstrapMsg m = new BootstrapMsg(msgEle.src);
				ConnMessageMiscs.m_writeDataBlock(m.flatten() , outStream);
			} else if (msgEle.type==ConnMessageTypes.ACK_MSG) {
				outStream.writeInt(((Integer) msgEle.load).intValue());
			}
			outStream.flush();
		}
		catch (IOException ioexp) {
			finish=true;
			log.info("Closing ConnHandlerOut "+destNode+"\n"+ioexp);
		}
		catch (Exception ex) {
			finish=true;
			log.info("Closing ConnHandlerOut "+destNode+"\n"+ex);
		}
		catch (java.lang.StackOverflowError er) {
			log.error("Stack Overflow:"+er);
		}
		//try
	} //SendTCPByte
	
	private void acceptConnection() throws IOException
	{
		socket.setSoTimeout(SOCKETTIMEOUT);
		outStream = new DataOutputStream(socket.getOutputStream());
	}
	
	
	private void establishConnection() throws	IOException {
		socket = new Socket(destNode.getIP(), destNode.getPort());
		socket.setSoTimeout(SOCKETTIMEOUT);
		outStream = new DataOutputStream(socket.getOutputStream());
	}
	
	
	private synchronized OutMsgElement getNextElement(){
		try{
			while (msgQueue.isEmpty() && !finish){
				wait();
			}
			if (msgQueue.size()==MAXSEND) 
				notify();
			return finish ? null : (OutMsgElement)msgQueue.remove(0);
		}
		catch (InterruptedException iexp) {
			end();
			return null;
		}
	}
	
	private synchronized void addElement(OutMsgElement ele){
		while (msgQueue.size()>=MAXSEND) {
			try {
				wait();
			} catch (Exception ex) {
				log.error( ex+"");
			}
		}
		msgQueue.add(ele);
		if (msgQueue.size()==1)
			notify();
	}
	
	public static double getDefaultRTT() { return DEFAULTRTT; }
	
	public static double getDefaultRTO() { return DEFAULTRTT*RTOFACTOR; }
	
	public double getRTT() { return rtt; }
	
	public double getRTO() { return rtt*RTOFACTOR; }
	
	
}//Sender class
